package com.sanerzone.smscenter.biz.listener;

import java.util.List;
import java.util.Map;

import com.google.common.base.Strings;
import com.google.common.collect.Maps;
import com.sanerzone.common.support.sequence.MsgId;
import com.sanerzone.smscenter.biz.cache.AccountCacheHelper;
import com.sanerzone.smscenter.biz.utils.GatewayHelper;
import com.sanerzone.smscenter.common.tools.JedisUtils;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.sanerzone.common.support.utils.FstObjectSerializeUtil;
import com.sanerzone.smscenter.biz.entity.DataEntity;
import com.sanerzone.smscenter.biz.message.SMSMTMessage;
import com.sanerzone.smscenter.biz.message.SMSURREQMessage;
import com.sanerzone.smscenter.biz.utils.DateHelper;
import com.sanerzone.smscenter.biz.utils.StringHelper;
import com.sanerzone.smscenter.biz.utils.TableNameHelper;
import com.sanerzone.smscenter.common.tools.MQHelper;
import com.sanerzone.smscenter.common.tools.SpringContextHelper;

@Service
public class INSMSMTListener implements MessageListenerConcurrently {

	public static Logger logger = LoggerFactory.getLogger(INSMSMTListener.class);
	
	@Autowired
	public MQHelper mQUtils;

	@Override
	public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
		SqlSession sqlSession = SpringContextHelper.getBean(SqlSessionFactory.class).openSession();
		
		int index =0;
		DataEntity<SMSMTMessage> dataEntity = null;
		SMSMTMessage smsMTMessage = null;
		SMSURREQMessage sMSURREQMessage = null;

		try {
			for (MessageExt msg : msgs) 
			{
				logger.info("mt listener recv message: topic:{}, tags:{}, msgid:{}, key:{}", msg.getTopic(), msg.getTags(),
						msg.getMsgId(), msg.getKeys());
				
				smsMTMessage = (SMSMTMessage)FstObjectSerializeUtil.read(msg.getBody());
//
				if (smsMTMessage.getUpdateOrInsert()!=null&&smsMTMessage.getUpdateOrInsert()==3){
					continue;
				}
				if (null != smsMTMessage && smsMTMessage.getSmsREQMessage() != null)
                {
					index++;
					dataEntity = new DataEntity<SMSMTMessage>(TableNameHelper.getSmsSendTable(smsMTMessage.getId()), smsMTMessage);
					if (smsMTMessage.getUpdateOrInsert()!=null&&smsMTMessage.getUpdateOrInsert() == 2) {
						//修改短信发送状态
						Map<String,Object> map = Maps.newHashMap();
						map.put("sendStatus",smsMTMessage.getSendStatus());
						map.put("smsContent",smsMTMessage.getSmsContent());
						map.put("id",smsMTMessage.getId());
						map.put( "phone",smsMTMessage.getPhone() );
						map.put("tableName",TableNameHelper.getSmsSendTable(smsMTMessage.getId()));
						sqlSession.update("com.sanerzone.smscenter.biz.mapper.SmsSendMapper.batchUpdateSmsSendStatus", map);
					} else {//不设置，就是插入
						sqlSession.insert("com.sanerzone.smscenter.biz.mapper.SmsSendMapper.insert", dataEntity);
						if (smsMTMessage.getNeedSplit()==1) {
							String[] phones = smsMTMessage.getPhone().split(",");
							dataEntity.getData().setNeedSplit( 0 );
							dataEntity.getData().setShowLevel( 1 );//单条显示
							for (int i = 0; i < phones.length; i++) {
								dataEntity.getData().setId(new MsgId().toString());
								index++;
								dataEntity.getData().setPhone(phones[i]);
								//分割出来的每条订单计费
								dataEntity.getData().setSmsSize( GatewayHelper.smsSize( dataEntity.getData().getSmsContent() ));
								sqlSession.insert("com.sanerzone.smscenter.biz.mapper.SmsSendMapper.insert", dataEntity);
								if(index % 500 == 0) {
									sqlSession.commit();
								}
							}
						}

					}
                    if(index % 500 == 0) {
                        sqlSession.commit();
                    }

//                    //如果是确认失败的 用户直接反冲余额
                    if (smsMTMessage.getSendStatus().startsWith( "F" )){
						String amountKey = AccountCacheHelper.getAmountKey("sms", smsMTMessage.getSmsREQMessage().getUserid());
						int smsSize=smsMTMessage.getSmsSize();
						JedisUtils.incrBy( amountKey,smsSize );
						logger.info( "用户id:{},反充结束,反充条数为{}",smsMTMessage.getSmsREQMessage().getUserid(),smsSize );
					}


                    //推送失败UR
                    if((smsMTMessage.getSendStatus().startsWith("F")&&
                    		StringHelper.registeredDelivery(smsMTMessage.getSmsREQMessage().getRegisteredDelivery()))){

						String phones=smsMTMessage.getPhone();
						String[] phoneArray=phones.split( "," );
						for (int i=0;i<phoneArray.length;i++){
							String phoneTemp=phoneArray[i];
							//发送失败并且以F开头而且是需要发送确认报告的就发送状态确认报告
							String time = DateHelper.formatCmppDate(smsMTMessage.getSubmitTime());
							sMSURREQMessage = new SMSURREQMessage();
							sMSURREQMessage.setAccid(smsMTMessage.getSmsREQMessage().getAccId());
							sMSURREQMessage.setBizid(smsMTMessage.getId());
							sMSURREQMessage.setDoneTime(time);
							sMSURREQMessage.setGateWayId(smsMTMessage.getGatewayId());
							sMSURREQMessage.setMsgid(smsMTMessage.getId());
							sMSURREQMessage.setPhone(phoneTemp);
							sMSURREQMessage.setSourceGateWayId(smsMTMessage.getSmsREQMessage().getSourceGateWayId());
							sMSURREQMessage.setSpnumber(smsMTMessage.getSmsREQMessage().getSendnumber());
							sMSURREQMessage.setStat(smsMTMessage.getSendStatus());
							sMSURREQMessage.setSubmitTime(time);
							sMSURREQMessage.setTaskid(smsMTMessage.getSmsREQMessage().getTaskid());
							sMSURREQMessage.setUserid(smsMTMessage.getSmsREQMessage().getUserid());
							sMSURREQMessage.setReserve( smsMTMessage.getFailReason() );
							mQUtils.sendSmsUR(smsMTMessage.getId(), sMSURREQMessage);


						}

                    }
                }
				else
				{
					logger.error("INSMSMT-ERR：msgid:{}, key:{}", msg.getMsgId(), msg.getKeys());
				}
			}
			
			sqlSession.commit();
		} 
		catch (Exception e) 
		{
			logger.error("INSMSMT-ERR", e);
		}
		finally
		{
			if (sqlSession != null)
            {
                sqlSession.close();
				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
		}
			
		return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
	}

}
